/*
 * st-sync-redis.c
 *
 *  Created on: May 18, 2017
 *      Author: zhangzm
 */
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <inttypes.h>
#include <stdbool.h>
#include <errno.h>

#include "st.h"
#include "hiredis.h"
#include "async.h"
#include "st-redis.h"
#include "networking.h"
#include "ursax.h"
#include "log.h"
#include "utcache.h"

static void redisContextCloseFd(redisContext *c)
{
	if (c && c->fd >= 0) {
		close(c->fd);
		c->fd = -1;
	}
}

static int redisCheckSocketError(redisContext *c)
{
	int err = 0;
	socklen_t errlen = sizeof(err);

	if (getsockopt(c->fd, SOL_SOCKET, SO_ERROR, &err, &errlen) == -1) {
		LOG_ERROR("redis socket getsockopt error");
		return REDIS_ERR;
	}
	if (err) {
		LOG_ERROR("redis socket getsockopt error1");
		return REDIS_ERR;
	}
	return REDIS_OK;
}

static int redisContextWaitReady(redisContext *c, long msec)
{
	int res = wait_socket_ready(c->fd, false, msec * 1000);
	if(res < 0){
		redisContextCloseFd(c);
		return REDIS_ERR;
	}

	if (redisCheckSocketError(c) != REDIS_OK){
		return REDIS_ERR;
	}

	return REDIS_OK;
}

struct RedisSTAttach* redis_sync_connect(char* ip, uint16_t port)
{
	redisContext *c = redisConnectNonBlock(ip, port);
	if(!c){
		LOG_ERROR("error redisConnectNonBlock");
		return NULL;
	}
	int ret = redisContextWaitReady(c, 1500);
	if(ret == REDIS_ERR){
		LOG_ERROR("error redisContextWaitReady");
		redisFree(c);
		return NULL;
	}
	struct RedisSTAttach* ra = (struct RedisSTAttach*)zalloc(sizeof(*ra));
	ra->sync_context = c;
	return ra;
}

struct RedisSTAttach* redis_sync_connect_elastic(char* ip, uint16_t port)
{
	struct RedisSTAttach *attach = NULL;
	MAKE_FD_ELASTIC(
		attach = redis_sync_connect(ip, port),
		"FAILED TO CONNECT TO REDIS SERVER", NULL
	);

	return attach;
}

struct redisReply* redisSyncCommandv(struct RedisSTAttach *ac, const char* format, va_list arglist)
{
	int ret = redisvAppendCommand(ac->sync_context, format, arglist);
	if(ret < 0){
		return NULL;
	}
	int wdone = 0;
	do {
		if (redisBufferWrite(ac->sync_context, &wdone) == REDIS_ERR)
			return NULL;
		if(!wdone){
			LOG_INFO("redisBufferWrite failed, continue");
			st_usleep(10000);
		}
	} while (!wdone);

	void* reply = NULL;
	do {
		int res = wait_socket_ready(ac->sync_context->fd, true, 60* 1000 * 1000);
		if(res < 0){
			return NULL;
		}
		if (redisBufferRead(ac->sync_context) == REDIS_ERR){
			LOG_ERROR("error redisBufferRead error %m");
			return NULL;
		}
		if (redisGetReplyFromReader(ac->sync_context, &reply) == REDIS_ERR){
			LOG_ERROR("error redisGetReplyFromReader error %m");
			return NULL;
		}
		if(reply == NULL){
			LOG_INFO("redisGetReplyFromReader return NULL, continue");
		}
	} while(reply == NULL);
	return (struct redisReply*)reply;
}

struct redisReply* redisSyncCommand(struct RedisSTAttach *ac, const char* format, ...)
{
	va_list arglist;
	va_start( arglist, format );
	struct redisReply* ret = redisSyncCommandv(ac, format, arglist);
	va_end( arglist );
	return ret;
}

void st_redis_freeObject(struct RedisSTAttach* ra, void* obj)
{

}

void redis_sync_free(struct RedisSTAttach* ra)
{
	redisFree(ra->sync_context);
	free(ra);
}
